package com.geezdata.cps.cqrs.event.publish.impl;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import javax.jms.JMSException;
import javax.jms.TextMessage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.core.JmsTemplate;

import com.geezdata.cps.base.utils.JsonUtil;
import com.geezdata.cps.cqrs.event.DelayEvent;
import com.geezdata.cps.cqrs.event.Event;
import com.geezdata.cps.cqrs.event.annotation.EventPublisher;
import com.geezdata.cps.cqrs.event.enums.EventSource;
import com.geezdata.cps.cqrs.event.listener.container.ListenerContainer;
import com.geezdata.cps.cqrs.event.publish.AbstractPublisher;


@EventPublisher(source = EventSource.MQ)
public class JmsPublisher extends AbstractPublisher {
	private static Logger logger = LoggerFactory.getLogger(JmsPublisher.class);
	
    private static final int MAX_CONSUMER = 5;
    public static final String AMQ_SCHEDULED_DELAY = "AMQ_SCHEDULED_DELAY";    
    private ExecutorService consumer = Executors.newFixedThreadPool(MAX_CONSUMER);

    private JmsTemplate jmsTemplate;

    public JmsPublisher(ListenerContainer listenerContainer, JmsTemplate jmsTemplate) {
        super(listenerContainer);

        if (jmsTemplate == null) {
            throw new IllegalArgumentException("JmsTemplate can't be null.");
        }
        this.jmsTemplate = jmsTemplate;
    }

    @Override
    public <T> void publish(Event<T> event) {
        consumer.execute(() -> {
            if (event == null) {
                return;
            }
            try {
                jmsTemplate.send(event.getName(), session -> {
                    TextMessage tm;
                    if (event.getPayLoad() instanceof String) {
                        tm = session.createTextMessage((String) event.getPayLoad());
                    } else {
                        tm = session.createTextMessage(JsonUtil.toJson(event.getPayLoad()));
                    }
                    handleIfDelay(event, tm);
                    return tm;
                });
            } catch (Exception e) {
                logger.error("JmsPublisher catch exception. {}", e);
            }
        });
    }

    @SuppressWarnings("rawtypes")
	private void handleIfDelay(Event event, TextMessage tm) throws JMSException {
        if (event instanceof DelayEvent) {
            DelayEvent de = (DelayEvent) event;
            tm.setLongProperty(AMQ_SCHEDULED_DELAY, de.getDelayTimestamp());
        }
    }

    public void stop() {
        try {
            consumer.shutdown();
            if (!consumer.awaitTermination(20, TimeUnit.SECONDS)) {
                logger.error("Await termination timeout.");
            }
        } catch (InterruptedException e) {
            logger.error("Catch interrupted exception. ", e);
        }
    }
}
